package com.laogao.lesson034.mq.consumer;

import cn.hutool.json.JSONUtil;
import com.laogao.lesson034.common.BusinessExceptionUtils;
import com.laogao.lesson034.mq.dto.Msg;
import com.laogao.lesson034.mq.po.MsgConsumePO;
import com.laogao.lesson034.mq.po.SequentialMsgConsumePositionPO;
import com.laogao.lesson034.mq.po.SequentialMsgQueuePO;
import com.laogao.lesson034.mq.service.ISequentialMsgConsumePositionService;
import com.laogao.lesson034.mq.service.ISequentialMsgQueueService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * 幂等消费者，子类集成该类，便拥有了幂等消费 && 消费失败衰减式重试的能力
 * <b>description</b>： Java进阶、性能优化实战案例100讲，微信公众号:程序员恰恰，源码 & 文档 & 技术支持，请加个人微信号：it3306 <br>
 * <b>time</b>：2025/3/30 15:16 <br>
 * <b>author</b>：ready lao_gao
 */
@Slf4j
public abstract class AbstractSequentialMsgConsumer<B, M extends Msg<B>> extends AbstractRetryConsumer<B, M> {

    @Autowired
    protected ISequentialMsgConsumePositionService sequentialMsgConsumeInfoService;

    @Autowired
    protected ISequentialMsgQueueService sequentialMsgQueueService;


    @Override
    protected void consume(Message message, M msg, MsgConsumePO msgConsumerPO) {
        //1、消息过来后，会将其放到db中搞的一个队列中进行排队
        this.pushQueue(message, msg);

        //2、从db中队列表中拉取消息进行消费（循环拉取最小的一条记录，看看是不是要消费的记录，如果是，则进行消费）
        this.pullMsgFromQueueConsume(message, msg);
    }

    /**
     * 从db中消息排队表拉取消息进行消费
     *
     * @param message
     * @param msg
     */
    protected void pullMsgFromQueueConsume(Message message, M msg) {
        String groupId = msg.getSequentialMsgGroupId();
        String queueName = this.getQueueName(message);
        String lockKey = String.format("consumeOrderMessage:%s:%s", groupId, queueName);

        //加分布式锁
        boolean lockResult = this.distributeLock.accept(lockKey, lk -> {
            //从db中队列表中拉取消息进行消费（循环拉取最小的一条记录，和当前消费位置对比下？看看是不是要消费的记录，如果是，则进行消费）
            while (true) {
                //从队列中拿到第一条消息进行消费
                SequentialMsgQueuePO firstMsg = this.sequentialMsgQueueService.getFirst(groupId, queueName);
                //队列中没有消息，退出循环
                if (firstMsg == null) {
                    break;
                }
                //获取队列当前消费的位置
                SequentialMsgConsumePositionPO sequentialMsgConsumePositionPO = this.sequentialMsgConsumeInfoService.getAndCreate(groupId, queueName);

                //轮到自己消费了？
                //当前消息的编号恰好是当前消费位置的下一条消息的编号，此时可以进行消费操作，消费完成后更新消费位置并从队列中移除该消息。

                //重点:循环拉最小的编号，直到退出while
                if (firstMsg.getNumbering() == sequentialMsgConsumePositionPO.getConsumeNumbering() + 1) {
                    //消费,由子类实现
                    this.sequentialMsgConsume(this.getMsg(firstMsg.getMsgJson()));
                    this.transactionTemplate.executeWithoutResult(action -> {
                        //更新消费位置
                        sequentialMsgConsumePositionPO.setConsumeNumbering(firstMsg.getNumbering());
                        boolean update = this.sequentialMsgConsumeInfoService.updateById(sequentialMsgConsumePositionPO);
                        if (!update) {
                            throw BusinessExceptionUtils.businessException("系统繁忙，请稍后重试");
                        }
                        //从队列中移除消息
                        this.sequentialMsgQueueService.delete(firstMsg.getId());
                    });

                } else if (msg.getSequentialMsgNumbering() < sequentialMsgConsumePositionPO.getConsumeNumbering()) {
                    //编号小于当前位置，说明已经消费过了，从队列中移除消息
                    //表明当前消息的编号小于当前消费位置，说明这条消息已经被消费过了，直接从队列中移除即可。
                    this.sequentialMsgQueueService.delete(firstMsg.getId());
                } else {
                    //当以上两个条件都不满足时，说明当前消息的编号大于当前消费位置的下一条消息的编号，也就是这条消息还未轮到被消费。在这种情况下，退出循环，等待后续消费位置更新后再尝试消费。
                    //还未轮到自己，退出循环
                    break;
                }
            }
        });
        /**
         * 加分布式锁失败，抛个异常，触发重试,抢不到锁的走失败重试逻辑
         * 极端情况:
         * 如果现在发了两条消息，消息1和消息2,后面的消息2先到，然后后面消息2获取分布式锁成功了，而第一个消息1此时又来了，
         * 此时消息1它会获取分布式锁失败。
         * 如果此时不抛异常，消息1就从MQ中被干掉了。被干掉之后，这个消息1后面就不会再消费了，就无法触发消费了。
         */
        if (!lockResult) {
            throw BusinessExceptionUtils.businessException("顺序消息消费加锁失败");
        }
    }

    /**
     * 将消息先压入队列
     *
     * @param message
     * @param msg
     */
    private void pushQueue(Message message, M msg) {
        String groupId = msg.getSequentialMsgGroupId();
        String queueName = this.getQueueName(message);
        Long numbering = msg.getSequentialMsgNumbering();
        this.sequentialMsgQueueService.push(groupId, numbering, queueName, JSONUtil.toJsonStr(msg));
    }

    protected abstract void sequentialMsgConsume(M msg);
}
